package com.shawn.integeration.rocketmq;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * rocketmq 消息生产者
 *
 * @author: shawn
 * @Date: 2024/1/31 16:27
 * @Version 1.0.0
 */
@Slf4j
@Component
public class RocketMqProducer {
    @Resource
    private RocketMQTemplate rocketmqTemplate;

    /**
     * 正常发送
     *
     * @param topic 主题
     * @param mqMsg 待发送消息对象
     */
    public void send(String topic, BaseMqMsg<?> mqMsg) {
        try {
            String msg = JSONUtil.toJsonStr(mqMsg);
            log.info("发送MQ消息 topic:{} msg:{}", topic, msg);
            rocketmqTemplate.convertAndSend(topic, msg);
        } catch (Exception e) {
            log.error("发送MQ消息失败 topic:{} msg:{}", topic, JSONUtil.toJsonStr(mqMsg), e);
            // 发送失败,一般需要进行消息补偿
        }
    }

    /**
     * 延迟发送
     *
     * @param topic
     * @param mqMsg
     * @param delayTimeLevel
     */
    public void delaySend(String topic, BaseMqMsg<?> mqMsg, int delayTimeLevel) {
        try {
            String msg = JSONUtil.toJsonStr(mqMsg);
            log.info("延迟发送MQ消息 topic:{} msg:{}", topic, msg);
            rocketmqTemplate.syncSend(topic, MessageBuilder.withPayload(mqMsg).build(), 1000, delayTimeLevel);
        } catch (Exception e) {
            log.error("延迟发送MQ消息失败 topic:{} msg:{}", topic, JSONUtil.toJsonStr(mqMsg), e);
            // 发送失败,一般需要进行消息补偿
        }
    }
}
